package plgnet

import (
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/gohttp"
	"dgo/goutils/utils"
	"fmt"
	"gitee.com/ymofen/gobase/golog"
	"gitee.com/ymofen/supervalue"
	"golang.org/x/net/websocket"
	"net/http"
	"strings"
	"sync/atomic"
)

/*
推荐使用

	2023-07-07 10:45:36
*/
type PlgHttpWsServV2023 struct {
	subsessionid uint32
	online       int32
	PlgBase
	msglst *utils.SyncCycleList

	debugmode  int8
	serviceUrl string

	// 订阅发送
	workers *utils.MultiGroupTask

	// 同步分发
	wsex *utils.Subscribe

	monitor_recv *utils.MonitorSpeedSizeRec
	monitor_send *utils.MonitorSpeedSizeRec
}

/*
	{
	  "id":"wsserv",
	  "type":"net.ws.serv.v2023",
	  "conf":{
		"serviceurl":"/ws/service"
	   }
	}
*/
func (this *PlgHttpWsServV2023) ReloadConf(conf *wrapper.SuperValue) error {
	if conf == nil {
		return nil
	}

	if this.msglst == nil {
		this.msglst = utils.NewSyncCycleList()
	}

	if this.wsex == nil {
		this.wsex = utils.NewSubscribe()
	}

	if this.workers == nil {
		this.workers = utils.NewMultiGroupTask()
		this.workers.ConfigWorkType(2)
		this.workers.ConfigMinWork(1)
		this.workers.ConfigMaxWork(32)
		this.workers.Start()
	}

	if this.monitor_recv == nil {
		this.monitor_recv = utils.NewMonitorSpeedSizeRec()
		utils.Monitor.AddMonitor(this.monitor_recv)
	}

	if this.monitor_send == nil {
		this.monitor_send = utils.NewMonitorSpeedSizeRec()
		utils.Monitor.AddMonitor(this.monitor_send)
	}

	conf.CheckExistsStrFunc("serviceurl", func(strv string) {
		this.ConfigServiceURL(strv)
	})

	if len(this.serviceUrl) == 0 {
		golog.Warnf("没有启动Ws服务, 没有配置serviceurl参数")
	}

	return nil
}

func (this *PlgHttpWsServV2023) Close() (err error) {
	utils.Monitor.RemoveMonitorRec(this.monitor_recv)
	utils.Monitor.RemoveMonitorRec(this.monitor_send)
	this.workers.Stop()
	this.PlgBase.Close()
	return
}

func (this *PlgHttpWsServV2023) onHttpHandle(conn *websocket.Conn) {

	if utils.GoFunCatchException {
		defer utils.PanicHandlerWithDebugString("WsHttpService-onHttpHandle")
	}

	atomic.AddInt32(&this.online, 1)
	defer func() {
		atomic.AddInt32(&this.online, -1)
	}()

	sn := atomic.AddUint32(&this.subsessionid, 1)
	sessionid := fmt.Sprintf("%s%d", this.PlugId, sn)

	subchannels := make(map[string]int8)
	defer func() {
		for k, v := range subchannels {
			if v == 1 {
				this.wsex.Unsub(sessionid, k)
			}
		}
	}()

	var sendfunc = func(msgtye byte, msg []byte) {
		this.workers.PostTaskFunc(sessionid, func() { // 排队发送要不然会乱
			err := SendWsMessage(conn, msgtye, msg)
			if err != nil {
				conn.Close()
			} else {
				this.monitor_send.DeltaBuf(msg)
			}
		})
	}

	var subfunc = func(topic string) {
		if len(topic) == 0 {
			return
		}
		if subchannels[topic] == 1 {
			return
		}
		this.wsex.Sub(sessionid, topic, func(id, topic string, args ...interface{}) (ok bool) {
			if buf, ok := args[0].([]byte); ok {
				sendfunc(WS_MESSAGE_BIN, buf)
			} else if str, ok := args[0].(string); ok {
				sendfunc(WS_MESSAGE_TEXT, []byte(str))
			}
			return true
		})
		subchannels[topic] = 1
	}

	str := conn.Request().URL.Query().Get("channels")
	if len(str) > 0 {
		strs := strings.Split(str, ",")
		for i := 0; i < len(strs); i++ {
			subfunc(utils.Trim(strs[i]))
		}
	}

	for {
		msgtype, msg, err := ReceiveMessage(conn)
		if err != nil {
			break
		}
		this.monitor_recv.DeltaBuf(msg)
		json, err := wrapper.NewSVFrom(msg, false)
		if err != nil {
			if this.debugmode == 1 {
				this.LogWarnMessage("[%s]非法的数据, msgtype:%d, msg:%s, err:%s", conn.RemoteAddr(), msgtype, utils.BufToHexStr(msg, 0, ""), err.Error())
			}
			conn.Close()
			break
		}

		topic := json.StringByName(WS_TOPIC_KEY, "")
		if topic == WS_HTTP_CHANNEL_CMD {
			cmd := json.StringByName("cmd", "")
			if cmd == WS_HTTP_CMD_subscribe {
				str := json.StringByName("channels", "")
				cnt := 0
				if len(str) > 0 {
					strs := strings.Split(str, ",")
					for i := 0; i < len(strs); i++ {
						subfunc(utils.Trim(strs[i]))
						cnt++
					}
				}

				sendfunc(WS_MESSAGE_TEXT, utils.BuildJSONBytes("code", 0, "cmd", cmd, "channels", str, "n", cnt, "client-addr", conn.RemoteAddr()))
			} else if cmd == WS_HTTP_CMD_unsubscribe {
				str := json.StringByName("channels", "")
				cnt := 0
				if len(str) > 0 {
					strs := strings.Split(str, ",")
					for i := 0; i < len(strs); i++ {
						topicstr := utils.Trim(strs[i])
						if len(topicstr) > 0 && subchannels[topicstr] == 1 {
							this.wsex.Unsub(sessionid, topicstr)
							cnt++
						}
					}
				}
				sendfunc(WS_MESSAGE_TEXT, utils.BuildJSONBytes("code", 0, "cmd", cmd, "channels", str, "n", cnt, "client-addr", conn.RemoteAddr()))
			}
		} else if topic == "__ECHO" {
			sendfunc(WS_MESSAGE_TEXT, msg)
		} else {
			cnt := this.wsex.Pub(topic, 0, json.AsJSONString(false))
			sendfunc(WS_MESSAGE_TEXT, utils.BuildJSONBytes("code", 0, "topic", topic, "n", cnt))
		}
	}

}

func (this *PlgHttpWsServV2023) ConfigServiceURL(strv string) {
	if this.serviceUrl != strv {
		if len(this.serviceUrl) > 0 {
			gohttp.UnHandle(this.serviceUrl)
		}
		this.serviceUrl = strv
		if len(this.serviceUrl) > 0 {
			wshd := websocket.Server{
				Handshake: func(config *websocket.Config, req *http.Request) (err error) {
					config.Origin, err = websocket.Origin(config, req)
					return nil // Origin 非法时不断开
				},
				Handler: this.onHttpHandle,
			}
			//wshd := websocket.Handler(this.onHttpHandle)
			gohttp.Handle(this.serviceUrl, wshd)
		}
	}
}

/*
直接推送
*/
func (this *PlgHttpWsServV2023) SendKVs(topic string, kv ...interface{}) error {
	if len(kv) == 1 {
		if json, ok := kv[0].(*wrapper.SuperValue); ok {
			json.SetKeyValue(WS_TOPIC_KEY, topic)
			this.wsex.Pub(topic, 0, json.AsJSONString(false))
		} else if str, ok := kv[0].(string); ok {
			this.wsex.Pub(topic, 0, str)
		}
		return nil
	}

	json := wrapper.NewSVObject()
	json.SetKeyValue(WS_TOPIC_KEY, topic)
	json.SetKeyValues(kv...)
	this.wsex.Pub(topic, 0, json.AsJSONString(false))
	return nil
}

func (this *PlgHttpWsServV2023) DispatchData(channelid string, sender interface{}, data interface{}, tag interface{}) (err error) {
	err = this.SendKVs(channelid, data)
	return err
}

/*
响应是在接收线程中, 不要堵塞
*/
func (this *PlgHttpWsServV2023) SubscribeChannel(channelid string, id interface{}, cb utils.SubscribeCallback) {
	this.wsex.Sub(fmt.Sprintf("%p", id), channelid, func(subid, topic string, args ...interface{}) (ok bool) {
		isbreak := false
		cb(nil, topic, id, args[0], nil, &isbreak)
		return true
	})
}

func (this *PlgHttpWsServV2023) UnsubscribeChannel(channelid string, id interface{}) bool {
	return this.wsex.Unsub(fmt.Sprintf("%p", id), channelid)
}

func (this *PlgHttpWsServV2023) DoStatusRequestRequest(args ...interface{}) (resp interface{}) {
	if len(args) == 0 {
		return fmt.Errorf("非法的参数")
	}
	json, _ := wrapper.NewSVFrom(args[0], false)

	if json != nil {
		itype := json.IntByName("type", 0)
		if itype == 99 {
			json := wrapper.NewSVObject()
			json.SetKeyValues("now", utils.NowString())
			return json
		} else if itype == 98 {
			return this.msglst
		} else if itype == 1 {
			json := wrapper.NewSVArray()
			//searchval := json.StringByName("searchval", "")
			//sn := 0
			//maxn := json.IntByName("n", 100)
			//this.lst.ListRangeV2(func(key, val interface{}) bool {
			//	pojo := val.(*VrsSitePojo)
			//	ok := true
			//	if len(searchval) > 0 {
			//		ok = strings.Contains(pojo.id, searchval)
			//	}
			//	return ok
			//}, func(k0, v0, k1, v1 interface{}) bool {
			//	p0 := v0.(*VrsSitePojo)
			//	p1 := v1.(*VrsSitePojo)
			//	return p0.id < p1.id
			//}, func(key, val interface{}) bool {
			//	sn++
			//	pojo := val.(*VrsSitePojo)
			//	itm := json.AppendObject()
			//	itm.SetKeyValue("sn", sn)
			//	itm.SetKeyValue("id", pojo.id)
			//	return sn < maxn
			//})

			return json
		} else {
			//resp = this.DoRequest(args...)
			//if resp != nil {
			//	return resp
			//}
		}
	}
	var sb utils.BytesBuilder
	PrjHttpUtils.AppendIndexBody(&sb, "status", "status?type=99")
	PrjHttpUtils.AppendIndexBody(&sb, "msglist", "status?type=98&n=100&searchval=")
	PrjHttpUtils.AppendIndexBody(&sb, "lst", "status?type=1&n=100&searchval=")
	return sb.Bytes()

}

func (this *PlgHttpWsServV2023) LogInfoMessage(s string, args ...interface{}) {
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_INFO, 1, s, args...)
}

func (this *PlgHttpWsServV2023) LogWarnMessage(s string, args ...interface{}) {
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_WARN, 1, s, args...)
}

func CreatePlgHttpWsServV2023Func(args ...interface{}) (rval interface{}, err error) {
	obj := &PlgHttpWsServV2023{}

	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.PlugId = str
		}
	}
	rval = obj
	return
}
